package com.dec.kks.etl.producer;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;

public class DECConfigUtil {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerMain.class);

    public static Map<String, Object> loadConfig(String confPath, int mode) throws Exception {
        Map<String, Object> kafkaConfg = new HashMap<>();
        Properties props = new Properties();
        try {
            String filename = "producer.properties";
            if (mode == 0) {
                filename = "consumer.properties";
            }
            if (StringUtils.isNotBlank(confPath) && !confPath.endsWith(File.separator)) {
                confPath = confPath + File.separator + filename;
            } else {
                confPath = filename;
            }
            props.load(new FileInputStream(confPath));
            Iterator<Map.Entry<Object, Object>> it = props.entrySet().iterator();
            while (it.hasNext()) {
                Map.Entry<Object, Object> tmp = it.next();
                String key = String.valueOf(tmp.getKey());
                Object value = tmp.getValue();
                kafkaConfg.put(key, value);
                LOGGER.info("kafka producer config", kafkaConfg);
            }
            return kafkaConfg;
        } catch (IOException e) {
            LOGGER.error("配置文件加载路径不正确!" + confPath, e);
            throw new Exception("配置文件加载路径不正确!" + confPath, e);
        }
    }
}
